package boot.spring.controller;

import boot.spring.model.entity.Intersection;
import boot.spring.model.po.TailMessageDataV2;
import boot.spring.model.po.TailMessageV2;
import boot.spring.model.po.TrialClient;
import boot.spring.service.IntersectionService;
import boot.spring.service.RedisTrailSubscriptionService;
import cn.hutool.extra.spring.SpringUtil;
import com.alibaba.fastjson.JSON;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import javax.websocket.*;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

@ServerEndpoint("/icv/perception")
@Component
public class TrailWebSocketServer {

    private static final Logger logger = LoggerFactory.getLogger(TrailWebSocketServer.class);

    private final IntersectionService intersectionService = SpringUtil.getBean(IntersectionService.class);
    private final RedisTrailSubscriptionService redisTrailSubscriptionService = SpringUtil.getBean(RedisTrailSubscriptionService.class);
    private final ObjectMapper objectMapper = new ObjectMapper();

    /**
     * 与某个客户端的连接会话，需要通过它来给客户端发送数据
     */
    private Session session;

    /**
     * 静态变量，用来记录当前在线连接数。应该把它设计成线程安全的。
     */
    private static AtomicInteger onlineNum = new AtomicInteger();

    /**
     * concurrent包的线程安全Set，用来存放每个客户端对应的WebSocketServer对象。
     */
    private static ConcurrentHashMap<String, Session> sessionPools = new ConcurrentHashMap<>();

    public TrailWebSocketServer() {
    }

    /**
     * 建立连接成功调用
     *
     * @param session
     */
    @OnOpen
    public void onOpen(Session session) {
        this.session = session;
        sessionPools.put(session.getId(), session);
        addOnlineCount();
        logger.info("{}加入webSocket！当前人数为:{}", session.getId(), onlineNum);
    }

    @OnMessage
    public void onMessage(String message, Session session) throws IOException {
        logger.info("---------------------成功与{}建立连接---------------------", session.getId());
        TailMessageV2 msg = JSON.parseObject(message, TailMessageV2.class);
        List<TailMessageDataV2> features = msg.getFeatures();
        Set<String> intersectionIdSet = new HashSet<>();
        if (features != null && !features.isEmpty()) {
            for (TailMessageDataV2 feature : features) {
                String geometry = feature.getGeometry();
                JsonNode coordinates = objectMapper.readValue(geometry, JsonNode.class).get("coordinates");
                double[][] result = new double[coordinates.size()][2];
                for (int i = 0; i < coordinates.size(); i++) {
                    JsonNode point = coordinates.get(i);
                    result[i][0] = point.get(0).asDouble();
                    result[i][1] = point.get(1).asDouble();
                }
                String wkt = convertToWkt(result);
                List<Intersection> intersectionList = intersectionService.getIntersectionsByPolygon(wkt);
                List<String> intersectionIdList = intersectionList.stream().map(Intersection::getDeviceId).collect(Collectors.toList());
                if (!intersectionIdList.isEmpty()) {
                    intersectionIdSet.addAll(intersectionIdList);
                }
            }
        }
        if (!intersectionIdSet.isEmpty()) {
            List<String> intersectionIdList = new ArrayList<>(intersectionIdSet);
            TrialClient trialClient = new TrialClient(session.getId(), intersectionIdList, this);
            redisTrailSubscriptionService.subscribe(trialClient);
        }
    }

    public static String convertToWkt(double[][] coordinates) {
        StringBuilder wktBuilder = new StringBuilder("POLYGON ((");
        for (int i = 0; i < coordinates.length; i++) {
            wktBuilder.append(coordinates[i][0])
                    .append(" ")
                    .append(coordinates[i][1]);
            if (i < coordinates.length - 1) {
                wktBuilder.append(", ");
            }
        }
        wktBuilder.append("))");
        return wktBuilder.toString();
    }

    /**
     * 发送消息
     *
     * @param message
     * @throws IOException
     */
    public synchronized void sendMessage(String message) throws IOException {
        if (this.session.isOpen()) {
            this.session.getBasicRemote().sendText(message);
        }
    }

    //错误时调用
    @OnError
    public void onError(Session session, Throwable throwable) {
        logger.info("---------------------与{}的连接发生错误---------------------", session.getId());
        redisTrailSubscriptionService.unsubscribe(session.getId());
        throwable.printStackTrace();
    }

    //关闭连接时调用
    @OnClose
    public void onClose(Session session) {
        sessionPools.remove(session.getId());
        subOnlineCount();
        redisTrailSubscriptionService.unsubscribe(session.getId());
        logger.info("{}断开webSocket连接！当前人数为:{}", session.getId(), onlineNum);

    }

    public static void addOnlineCount() {
        onlineNum.incrementAndGet();
    }

    public static void subOnlineCount() {
        onlineNum.decrementAndGet();
    }

    public static AtomicInteger getOnlineNumber() {
        return onlineNum;
    }

    public static ConcurrentHashMap<String, Session> getSessionPools() {
        return sessionPools;
    }
}
